const aedes = require('aedes')();
const server = require('net').createServer(aedes.handle);
const express = require('express');
const http = require('http');
const WebSocket = require('ws');
const sqlite3 = require('sqlite3').verbose();
const cors = require('cors');
const bodyParser = require('body-parser');
const moment = require('moment');
const path = require('path');

// 配置
const MQTT_PORT = 1883;
const WEB_PORT = 3000;
const WS_PORT = 8080;

// 创建Express应用
const app = express();
app.use(cors());
app.use(bodyParser.json());
app.use(express.static('public'));

// 数据库初始化
const dbPath = path.join(__dirname, 'mqtt_data.db');
const db = new sqlite3.Database(dbPath);

// 初始化数据库表
function initDatabase() {
    db.serialize(() => {
        // 创建设备表
        db.run(`CREATE TABLE IF NOT EXISTS devices (
            id INTEGER PRIMARY KEY AUTOINCREMENT,
            device_id TEXT UNIQUE,
            name TEXT,
            status TEXT DEFAULT 'offline',
            last_seen DATETIME,
            created_at DATETIME DEFAULT CURRENT_TIMESTAMP
        )`);

        // 创建消息日志表
        db.run(`CREATE TABLE IF NOT EXISTS message_logs (
            id INTEGER PRIMARY KEY AUTOINCREMENT,
            topic TEXT,
            payload TEXT,
            client_id TEXT,
            timestamp DATETIME DEFAULT CURRENT_TIMESTAMP
        )`);

        // 创建寄存柜数据表
        db.run(`CREATE TABLE IF NOT EXISTS boxes (
            id INTEGER PRIMARY KEY,
            phone TEXT,
            password TEXT,
            boxId TEXT,
            size TEXT,
            status INTEGER,
            enabled BOOLEAN,
            lockBoard INTEGER,
            lockNumber INTEGER,
            memberFlag INTEGER,
            synced BOOLEAN DEFAULT 0,
            updated_at DATETIME DEFAULT CURRENT_TIMESTAMP
        )`);

        // 创建操作记录表
        db.run(`CREATE TABLE IF NOT EXISTS records (
            id INTEGER PRIMARY KEY AUTOINCREMENT,
            boxId INTEGER,
            operationTime TEXT,
            operator TEXT,
            actionType TEXT,
            accessMethod TEXT,
            synced BOOLEAN DEFAULT 0,
            updated_at DATETIME DEFAULT CURRENT_TIMESTAMP
        )`);

        // 创建成员表
        db.run(`CREATE TABLE IF NOT EXISTS members (
            id TEXT PRIMARY KEY,
            name TEXT,
            password TEXT,
            synced BOOLEAN DEFAULT 0,
            updated_at DATETIME DEFAULT CURRENT_TIMESTAMP
        )`);

        // 创建管理员用户表
        db.run(`CREATE TABLE IF NOT EXISTS admin_users (
            id INTEGER PRIMARY KEY,
            username TEXT,
            password TEXT,
            role TEXT,
            synced BOOLEAN DEFAULT 0,
            updated_at DATETIME DEFAULT CURRENT_TIMESTAMP
        )`);

        // 清理重复的设备记录，合并aedes_开头的设备
        db.run(`DELETE FROM devices WHERE device_id LIKE 'aedes_%'`);
        
        console.log('数据库初始化完成');
        console.log('已清理重复的设备记录');
    });
}

// WebSocket服务器用于实时通信
const wss = new WebSocket.Server({ port: WS_PORT });

// 广播消息给所有WebSocket客户端
function broadcastToWebClients(data) {
    console.log(`📡 准备广播数据:`, data);
    let sentCount = 0;
    wss.clients.forEach(client => {
        if (client.readyState === WebSocket.OPEN) {
            client.send(JSON.stringify(data));
            sentCount++;
        }
    });
    console.log(`✅ 已广播到 ${sentCount} 个Web客户端`);
}

// MQTT事件处理
aedes.on('client', (client) => {
    console.log(`🔌 客户端连接: ${client.id}`);
    // 修复：获取客户端数量
    const clientCount = aedes.clients ? Object.keys(aedes.clients).length : 0;
    console.log(`📊 当前连接的客户端数量: ${clientCount}`);
    
    // 提取设备名称并分配设备编号
    let deviceName = client.id;
    if (client.id.startsWith('aedes_')) {
        // 简化逻辑，直接使用"HarmonyOS设备1"
        deviceName = 'HarmonyOS设备1';
        
        // 检查是否已存在该设备
        db.get(`SELECT device_id FROM devices WHERE device_id = ?`, [deviceName], (err, row) => {
            if (!err) {
                if (!row) {
                    // 设备不存在，插入新记录
                    db.run(`INSERT INTO devices (device_id, name, status, last_seen) 
                            VALUES (?, ?, 'online', datetime('now'))`, [deviceName, deviceName], function(err) {
                        if (err) {
                            console.error('❌ 数据库插入设备失败:', err);
                        } else {
                            console.log('✅ 新设备已添加到数据库');
                        }
                    });
                } else {
                    // 设备已存在，更新状态
                    db.run(`UPDATE devices SET status = 'online', last_seen = datetime('now') 
                            WHERE device_id = ?`, [deviceName], function(err) {
                        if (err) {
                            console.error('❌ 数据库更新设备状态失败:', err);
                        } else {
                            console.log('✅ 设备状态已更新到数据库');
                        }
                    });
                }
                
                // 广播到Web客户端
                console.log(`📡 广播客户端连接事件到 ${wss.clients.size} 个Web客户端`);
                broadcastToWebClients({
                    type: 'client_connected',
                    clientId: deviceName,
                    timestamp: moment().format('YYYY-MM-DD HH:mm:ss')
                });
                
                // 请求客户端上传初始数据库数据
                console.log(`📋 请求客户端 ${deviceName} 上传初始数据库数据`);
                aedes.publish({
                    topic: 'kaihong/devices/database/request',
                    payload: JSON.stringify({
                        type: 'REQUEST_INITIAL_UPLOAD',
                        timestamp: moment().format(),
                        message: '请上传初始数据库数据'
                    })
                });
            }
        });
        return; // 提前返回，避免重复执行下面的代码
    }
    
    // 更新设备状态 - 使用设备名称而不是完整的client.id
    db.run(`INSERT OR REPLACE INTO devices (device_id, name, status, last_seen) 
            VALUES (?, ?, 'online', datetime('now'))`, [deviceName, deviceName], function(err) {
        if (err) {
            console.error('❌ 数据库更新设备状态失败:', err);
        } else {
            console.log('✅ 设备状态已更新到数据库');
        }
    });
    
    // 广播到Web客户端
    console.log(`📡 广播客户端连接事件到 ${wss.clients.size} 个Web客户端`);
    broadcastToWebClients({
        type: 'client_connected',
        clientId: deviceName,
        timestamp: moment().format('YYYY-MM-DD HH:mm:ss')
    });
});

aedes.on('clientDisconnect', (client) => {
    console.log(`🔌 客户端断开: ${client.id}`);
    // 修复：获取客户端数量
    const clientCount = aedes.clients ? Object.keys(aedes.clients).length : 0;
    console.log(`📊 当前连接的客户端数量: ${clientCount}`);
    
    // 提取设备名称
    let deviceName = client.id;
    if (client.id.startsWith('aedes_')) {
        deviceName = 'HarmonyOS设备1';
    }
    
    // 检查是否还有其他相同设备的连接
    let hasOtherConnections = false;
    // 修复：使用 Map 的正确迭代方法
    if (aedes.clients && typeof aedes.clients === 'object') {
        for (let [clientId, clientObj] of Object.entries(aedes.clients)) {
            if (clientId !== client.id && clientId.startsWith('aedes_')) {
                hasOtherConnections = true;
                break;
            }
        }
    }
    
    // 只有在没有其他连接时才设置为离线
    if (!hasOtherConnections) {
        db.run(`UPDATE devices SET status = 'offline', last_seen = datetime('now') 
                WHERE device_id = ?`, [deviceName], function(err) {
            if (err) {
                console.error('❌ 数据库更新设备离线状态失败:', err);
            } else {
                console.log('✅ 设备离线状态已更新到数据库');
            }
        });
        
        // 广播到Web客户端
        console.log(`📡 广播客户端断开事件到 ${wss.clients.size} 个Web客户端`);
        broadcastToWebClients({
            type: 'client_disconnected',
            clientId: deviceName,
            timestamp: moment().format('YYYY-MM-DD HH:mm:ss')
        });
    } else {
        console.log('🔄 设备仍有其他连接，保持在线状态');
    }
});

aedes.on('publish', (packet, client) => {
    // 过滤系统消息，避免无限循环
    if (packet.topic.startsWith('$SYS/')) {
        return;
    }
    
    if (client) {
        // 对于数据库上传消息，显示简化日志
        if (packet.topic === 'kaihong/devices/database/upload') {
            const payloadStr = packet.payload.toString();
            console.log(`📨 收到数据库上传消息 - 客户端: ${client.id}, 数据大小: ${payloadStr.length} 字符`);
            
            // 尝试解析并显示统计
            try {
                const data = JSON.parse(payloadStr);
                const boxCount = data.boxes ? data.boxes.length : 0;
                const recordCount = data.records ? data.records.length : 0;
                const memberCount = data.members ? data.members.length : 0;
                const adminCount = data.adminUsers ? data.adminUsers.length : 0;
                console.log(`📊 数据统计: 寄存柜${boxCount}条, 记录${recordCount}条, 成员${memberCount}条, 管理员${adminCount}条`);
            } catch (e) {
                console.log(`⚠️ 数据解析失败，可能数据格式有误`);
            }
        } else {
            // 减少心跳消息的日志输出
            if (packet.topic === 'heart' || packet.topic === 'jieshou') {
                // 心跳消息不输出日志，保持静默
                return;
            } else {
                console.log(`📨 收到消息 - 主题: ${packet.topic}, 客户端: ${client.id}, 内容: ${packet.payload.toString()}`);
            }
        }
        
        // 记录消息日志
        db.run(`INSERT INTO message_logs (topic, payload, client_id) 
                VALUES (?, ?, ?)`, [packet.topic, packet.payload.toString(), client.id], function(err) {
            if (err) {
                console.error('❌ 消息日志记录失败:', err);
            } else {
                if (packet.topic !== 'kaihong/devices/database/upload') {
                    console.log('✅ 消息已记录到数据库');
                }
            }
        });
        
        // 处理特定主题的消息
        handleTopicMessage(packet.topic, packet.payload.toString(), client.id);
        
        // 广播到Web客户端（数据库上传消息不广播，避免数据量过大）
        if (packet.topic !== 'kaihong/devices/database/upload') {
            console.log(`📡 广播消息到 ${wss.clients.size} 个Web客户端`);
            broadcastToWebClients({
                type: 'message_received',
                topic: packet.topic,
                payload: packet.payload.toString(),
                clientId: client.id,
                timestamp: moment().format('YYYY-MM-DD HH:mm:ss')
            });
        }
    } else {
        // 处理来自服务器内部的消息（包括HTTP API发布的消息）
        if (packet.topic === 'heart' || packet.topic === 'jieshou') {
            // 心跳消息静默处理
            return;
        } else {
            console.log(`🏠 服务器内部消息 - 主题: ${packet.topic}, 内容: ${packet.payload.toString()}`);
            
            // 重要：处理服务器内部消息的主题
            handleTopicMessage(packet.topic, packet.payload.toString(), 'server-internal');
        }
    }
});

// 处理特定主题的消息
function handleTopicMessage(topic, payload, clientId) {
    switch (topic) {
        // 新的主题处理
        case 'cabinet/device/status':
            handleDeviceStatus(payload, clientId);
            break;
        case 'cabinet/device/response':
            handleDeviceResponse(payload, clientId);
            break;
        case 'cabinet/database/upload':
            handleDatabaseUpload(payload, clientId);
            break;
        case 'cabinet/device/heartbeat':
            handleDeviceHeartbeat(payload, clientId);
            break;
        case 'miniprogram/commands':
            handleMiniprogramCommands(payload, clientId);
            break;
        case 'miniprogram/register':
            handleMiniprogramRegister(payload, clientId);
            break;
        case 'miniprogram/heartbeat':
            handleMiniprogramHeartbeat(payload, clientId);
            break;
            
        // 兼容旧主题处理
        case 'heart':
            handleHeartbeat(payload, clientId);
            break;
        case 'kaihong/devices/properties/report':
            handleDeviceReport(payload, clientId);
            break;
        case 'kaihong/devices/database/upload':
            handleDatabaseUpload(payload, clientId);
            break;
        case 'kaihong/devices/database/request':
            handleDatabaseRequest(payload, clientId);
            break;
        case 'kaihong/devices/cmd/put':
            handleCommand(payload, clientId);
            break;
            
        default:
            console.log(`未处理的主题: ${topic}`);
    }
}

// 处理心跳消息
function handleHeartbeat(payload, clientId) {
    // 减少心跳日志输出，只在需要时打印
    // console.log(`💓 心跳信号 - 客户端: ${clientId}, 内容: ${payload}`);
    
    // 更新设备最后活跃时间
    db.run(`UPDATE devices SET last_seen = datetime('now') WHERE device_id = ?`, [clientId]);
    
    // 检查心跳消息中是否包含控制命令
    try {
        const heartData = JSON.parse(payload);
        if (heartData.command && heartData.command.action) {
            console.log(`🎮 心跳中包含控制命令:`, heartData.command);
            handleCommand(payload, clientId);
        }
    } catch (e) {
        // 如果解析失败，说明是普通心跳消息，继续正常处理
        // 不输出错误，因为大部分心跳消息都是简单字符串
    }
}

// ==================== 新的主题处理函数 ====================

// 处理设备状态上报
function handleDeviceStatus(payload, clientId) {
    console.log(`📊 设备状态上报 - 客户端: ${clientId}`);
    
    try {
        const statusData = JSON.parse(payload);
        console.log(`📊 设备状态:`, statusData);
        
        // 更新设备状态到数据库
        // TODO: 实现设备状态更新逻辑
        
        // 广播设备状态变化
        broadcastToWebClients({
            type: 'device_status_update',
            deviceId: statusData.deviceId,
            status: statusData.status,
            boxStatus: statusData.boxStatus,
            timestamp: moment().format('YYYY-MM-DD HH:mm:ss')
        });
        
    } catch (e) {
        console.error('❌ 设备状态数据解析失败:', e);
    }
}

// 处理设备响应
function handleDeviceResponse(payload, clientId) {
    console.log(`📨 设备响应 - 客户端: ${clientId}`);
    
    try {
        const responseData = JSON.parse(payload);
        console.log(`📨 设备响应:`, responseData);
        
        // 转发响应给小程序
        aedes.publish({
            topic: 'miniprogram/response',
            payload: JSON.stringify(responseData)
        });
        
        // 广播设备响应
        broadcastToWebClients({
            type: 'device_response',
            deviceId: responseData.deviceId,
            status: responseData.status,
            message: responseData.message,
            data: responseData.data,
            timestamp: moment().format('YYYY-MM-DD HH:mm:ss')
        });
        
    } catch (e) {
        console.error('❌ 设备响应数据解析失败:', e);
    }
}

// 处理设备心跳
function handleDeviceHeartbeat(payload, clientId) {
    // 减少心跳日志输出
    // console.log(`💓 设备心跳 - 客户端: ${clientId}`);
    
    // 更新设备最后活跃时间
    db.run(`UPDATE devices SET last_seen = datetime('now') WHERE device_id = ?`, [clientId]);
}

// 处理小程序指令
function handleMiniprogramCommands(payload, clientId) {
    console.log(`📲 小程序指令 - 客户端: ${clientId}`);
    
    try {
        const commandData = JSON.parse(payload);
        console.log(`📲 小程序指令:`, commandData);
        
        // 兼容不同的字段命名：targetDevice 或 target_device
        const targetDevice = commandData.targetDevice || commandData.target_device || 'HarmonyOS设备1';
        
        // 转发指令给设备，使用设备期望的ControlMessage格式
        const deviceCommand = {
            command: commandData.command,
            timestamp: commandData.timestamp || new Date().toISOString(),
            target_device: targetDevice,
            testMode: commandData.testMode || false
        };
        
        console.log(`📲 转发给设备的指令:`, deviceCommand);
        
        aedes.publish({
            topic: 'cabinet/control/commands',
            payload: JSON.stringify(deviceCommand)
        });
        
        console.log(`✅ 小程序指令已转发给设备: ${targetDevice}`);
        
    } catch (e) {
        console.error('❌ 小程序指令数据解析失败:', e);
    }
}

// 处理小程序心跳
function handleMiniprogramHeartbeat(payload, clientId) {
    // console.log(`💓 小程序心跳 - 客户端: ${clientId}`);
    
    // 更新小程序最后活跃时间
    db.run(`UPDATE devices SET last_seen = datetime('now') WHERE device_id = ?`, [clientId]);
}

// 生成消息ID
function generateMessageId() {
    return 'msg_' + Date.now() + '_' + Math.random().toString(36).substr(2, 9);
}

// ==================== 兼容旧主题处理函数 ====================

// 处理小程序注册
function handleMiniprogramRegister(payload, clientId) {
    console.log(`📱 小程序客户端注册: ${clientId}`);
    
    try {
        const registerData = JSON.parse(payload);
        console.log(`📱 小程序信息:`, registerData);
        
        // 将小程序作为特殊设备记录到数据库
        db.run(`INSERT OR REPLACE INTO devices (device_id, name, status, last_seen) 
                VALUES (?, ?, 'online', datetime('now'))`, 
                [registerData.clientId, '微信小程序客户端', registerData.clientId], function(err) {
            if (err) {
                console.error('❌ 小程序设备记录失败:', err);
            } else {
                console.log('✅ 小程序已注册为在线设备');
            }
        });
        
        // 广播小程序连接事件
        broadcastToWebClients({
            type: 'miniprogram_connected',
            clientId: registerData.clientId,
            userAgent: registerData.userAgent,
            version: registerData.version,
            timestamp: moment().format('YYYY-MM-DD HH:mm:ss')
        });
        
    } catch (e) {
        console.error('❌ 小程序注册数据解析失败:', e);
    }
}

// 处理设备上报
function handleDeviceReport(payload, clientId) {
    try {
        const data = JSON.parse(payload);
        console.log(`设备上报 - 客户端: ${clientId}`, data);
        
        // 这里可以根据上报的数据更新设备状态
        // 例如更新设备的属性信息等
        
    } catch (error) {
        console.error('解析设备上报数据失败:', error);
    }
}

// 处理数据库请求
function handleDatabaseRequest(payload, clientId) {
    try {
        const data = JSON.parse(payload);
        console.log(`数据库请求 - 类型: ${data.type}, 客户端: ${clientId}`);
        
        // 这里可以根据需要处理不同类型的数据库请求
        // 目前主要是日志记录，实际的请求逻辑在连接时已经处理
        
    } catch (error) {
        console.error('处理数据库请求失败:', error);
    }
}

// 处理数据库上传
function handleDatabaseUpload(payload, clientId) {
    try {
        const data = JSON.parse(payload);
        console.log(`数据库上传 - 类型: ${data.type}, 客户端: ${clientId}`);
        
        if (data.type === 'FULL_DATABASE_UPLOAD') {
            handleFullDatabaseUpload(data, clientId);
        } else if (data.type === 'INCREMENTAL_UPDATE') {
            handleIncrementalUpdate(data, clientId);
        }
        
    } catch (error) {
        console.error('处理数据库上传失败:', error);
    }
}

// 处理全量数据库上传
function handleFullDatabaseUpload(data, clientId) {
    console.log(`📋 [数据库上传] 开始处理来自客户端 ${clientId} 的全量数据库上传`);
    console.log(`📋 [数据库上传] 数据统计: 寄存柜${data.boxes?.length || 0}条, 记录${data.records?.length || 0}条, 成员${data.members?.length || 0}条, 管理员${data.adminUsers?.length || 0}条`);
    
    db.serialize(() => {
        db.run('BEGIN TRANSACTION');
        
        try {
            // 备份现有数据（可选）
            console.log(`📋 [数据库上传] 开始清空现有数据...`);
            db.run('DELETE FROM boxes');
            db.run('DELETE FROM records');
            db.run('DELETE FROM members');
            db.run('DELETE FROM admin_users');
            console.log(`📋 [数据库上传] 现有数据清空完成`);
            
            // 插入寄存柜数据
            if (data.boxes && data.boxes.length > 0) {
                console.log(`📋 [数据库上传] 开始插入 ${data.boxes.length} 条寄存柜数据...`);
                const stmt = db.prepare(`INSERT INTO boxes 
                    (id, phone, password, boxId, size, status, enabled, lockBoard, lockNumber, memberFlag, synced, updated_at) 
                    VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?, 1, datetime('now'))`);
                
                data.boxes.forEach((box, index) => {
                    stmt.run([box.id, box.phone, box.password, box.boxId, box.size, 
                             box.status, box.enabled, box.lockBoard, box.lockNumber, box.memberFlag]);
                    if ((index + 1) % 10 === 0) {
                        console.log(`📋 [数据库上传] 已插入 ${index + 1}/${data.boxes.length} 条寄存柜数据`);
                    }
                });
                stmt.finalize();
                console.log(`✅ [数据库上传] 成功插入 ${data.boxes.length} 条寄存柜数据`);
            }
            
            // 插入操作记录
            if (data.records && data.records.length > 0) {
                console.log(`📋 [数据库上传] 开始插入 ${data.records.length} 条操作记录...`);
                const stmt = db.prepare(`INSERT INTO records 
                    (boxId, operationTime, operator, actionType, accessMethod, synced, updated_at) 
                    VALUES (?, ?, ?, ?, ?, 1, datetime('now'))`);
                
                data.records.forEach((record, index) => {
                    stmt.run([record.boxId, record.operationTime, record.operator, 
                             record.actionType, record.accessMethod]);
                    if ((index + 1) % 20 === 0) {
                        console.log(`📋 [数据库上传] 已插入 ${index + 1}/${data.records.length} 条操作记录`);
                    }
                });
                stmt.finalize();
                console.log(`✅ [数据库上传] 成功插入 ${data.records.length} 条操作记录`);
            }
            
            // 插入成员数据
            if (data.members && data.members.length > 0) {
                console.log(`📋 [数据库上传] 开始插入 ${data.members.length} 条成员数据...`);
                const stmt = db.prepare(`INSERT INTO members (id, name, password, synced, updated_at) VALUES (?, ?, ?, 1, datetime('now'))`);
                data.members.forEach((member, index) => {
                    stmt.run([member.id, member.name, member.password]);
                    if ((index + 1) % 10 === 0) {
                        console.log(`📋 [数据库上传] 已插入 ${index + 1}/${data.members.length} 条成员数据`);
                    }
                });
                stmt.finalize();
                console.log(`✅ [数据库上传] 成功插入 ${data.members.length} 条成员数据`);
            }
            
            // 插入管理员用户数据
            if (data.adminUsers && data.adminUsers.length > 0) {
                console.log(`📋 [数据库上传] 开始插入 ${data.adminUsers.length} 条管理员数据...`);
                const stmt = db.prepare(`INSERT INTO admin_users (id, username, password, role, synced, updated_at) VALUES (?, ?, ?, ?, 1, datetime('now'))`);
                data.adminUsers.forEach((admin, index) => {
                    stmt.run([admin.id, admin.username, admin.password, admin.role]);
                    if ((index + 1) % 5 === 0) {
                        console.log(`📋 [数据库上传] 已插入 ${index + 1}/${data.adminUsers.length} 条管理员数据`);
                    }
                });
                stmt.finalize();
                console.log(`✅ [数据库上传] 成功插入 ${data.adminUsers.length} 条管理员数据`);
            }
            
            db.run('COMMIT');
            
            // 计算总记录数
            const totalRecords = (data.boxes?.length || 0) + (data.records?.length || 0) + 
                               (data.members?.length || 0) + (data.adminUsers?.length || 0);
            console.log(`🎉 [数据库上传] 全量数据库上传处理完成！总共处理 ${totalRecords} 条记录`);
            
            // 发送确认消息给客户端
            console.log(`📋 [数据库上传] 发送成功确认消息给客户端 ${clientId}`);
            aedes.publish({
                topic: 'kaihong/devices/database/response',
                payload: JSON.stringify({
                    type: 'FULL_UPLOAD_SUCCESS',
                    timestamp: moment().format(),
                    message: `全量数据上传成功，共处理 ${totalRecords} 条记录`,
                    totalRecords: totalRecords,
                    clientId: clientId
                })
            });
            
        } catch (error) {
            db.run('ROLLBACK');
            console.error(`❌ [数据库上传] 全量数据库上传失败:`, error);
            
            // 发送错误消息给客户端
            console.log(`📋 [数据库上传] 发送失败消息给客户端 ${clientId}`);
            aedes.publish({
                topic: 'kaihong/devices/database/response',
                payload: JSON.stringify({
                    type: 'FULL_UPLOAD_ERROR',
                    timestamp: moment().format(),
                    message: '全量数据上传失败: ' + error.message,
                    clientId: clientId
                })
            });
        }
    });
}

// 处理增量更新
function handleIncrementalUpdate(data, clientId) {
    db.serialize(() => {
        db.run('BEGIN TRANSACTION');
        
        try {
            // 更新寄存柜数据
            if (data.boxes && data.boxes.length > 0) {
                const stmt = db.prepare(`INSERT OR REPLACE INTO boxes 
                    (id, phone, password, boxId, size, status, enabled, lockBoard, lockNumber, memberFlag, synced, updated_at) 
                    VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?, 1, datetime('now'))`);
                
                data.boxes.forEach(box => {
                    stmt.run([box.id, box.phone, box.password, box.boxId, box.size, 
                             box.status, box.enabled, box.lockBoard, box.lockNumber, box.memberFlag]);
                });
                stmt.finalize();
                console.log(`增量更新 ${data.boxes.length} 条寄存柜数据`);
            }
            
            // 插入新的操作记录
            if (data.records && data.records.length > 0) {
                const stmt = db.prepare(`INSERT INTO records 
                    (boxId, operationTime, operator, actionType, accessMethod, synced, updated_at) 
                    VALUES (?, ?, ?, ?, ?, 1, datetime('now'))`);
                
                data.records.forEach(record => {
                    stmt.run([record.boxId, record.operationTime, record.operator, 
                             record.actionType, record.accessMethod]);
                });
                stmt.finalize();
                console.log(`增量插入 ${data.records.length} 条操作记录`);
            }
            
            // 更新成员数据
            if (data.members && data.members.length > 0) {
                const stmt = db.prepare(`INSERT OR REPLACE INTO members 
                    (id, name, password, synced, updated_at) VALUES (?, ?, ?, 1, datetime('now'))`);
                data.members.forEach(member => {
                    stmt.run([member.id, member.name, member.password]);
                });
                stmt.finalize();
                console.log(`增量更新 ${data.members.length} 条成员数据`);
            }
            
            // 更新管理员用户数据
            if (data.adminUsers && data.adminUsers.length > 0) {
                const stmt = db.prepare(`INSERT OR REPLACE INTO admin_users 
                    (id, username, password, role, synced, updated_at) VALUES (?, ?, ?, ?, 1, datetime('now'))`);
                data.adminUsers.forEach(admin => {
                    stmt.run([admin.id, admin.username, admin.password, admin.role]);
                });
                stmt.finalize();
                console.log(`增量更新 ${data.adminUsers.length} 条管理员数据`);
            }
            
            // 处理删除记录
            if (data.deletions && data.deletions.length > 0) {
                data.deletions.forEach(deletion => {
                    switch (deletion.type) {
                        case 'box':
                            db.run('DELETE FROM boxes WHERE id = ?', [deletion.id]);
                            break;
                        case 'member':
                            db.run('DELETE FROM members WHERE id = ?', [deletion.id]);
                            break;
                        case 'admin':
                            db.run('DELETE FROM admin_users WHERE id = ?', [deletion.id]);
                            break;
                    }
                });
                console.log(`处理 ${data.deletions.length} 条删除记录`);
            }
            
            db.run('COMMIT');
            console.log('增量数据更新处理完成');
            
            // 发送确认消息给客户端
            aedes.publish({
                topic: 'kaihong/devices/database/response',
                payload: JSON.stringify({
                    type: 'INCREMENTAL_UPDATE_SUCCESS',
                    timestamp: moment().format(),
                    message: '增量数据更新成功'
                })
            });
            
        } catch (error) {
            db.run('ROLLBACK');
            console.error('增量数据更新失败:', error);
            
            // 发送错误消息给客户端
            aedes.publish({
                topic: 'kaihong/devices/database/response',
                payload: JSON.stringify({
                    type: 'INCREMENTAL_UPDATE_ERROR',
                    timestamp: moment().format(),
                    message: '增量数据更新失败: ' + error.message
                })
            });
        }
    });
}

// 处理命令消息
function handleCommand(payload, clientId) {
    try {
        const command = JSON.parse(payload);
        console.log(`🎮 收到命令 - 客户端: ${clientId}`, command);
        
        // 处理柜门控制命令
        if (command.command && command.command.action) {
            const action = command.command.action;
            
            switch(action) {
                case 'open_door':
                    console.log(`🔓 开启柜门命令: 柜门${command.command.boxId}, 锁号${command.command.lockNumber}`);
                    // 更新数据库中的柜门状态为开启(2)
                    if (command.command.doorId) {
                        db.run(`UPDATE boxes SET status = 2, updated_at = datetime('now') WHERE id = ?`, 
                               [command.command.doorId], function(err) {
                            if (err) {
                                console.error('❌ 更新柜门状态失败:', err);
                            } else {
                                console.log(`✅ 柜门${command.command.boxId}状态已更新为开启`);
                            }
                        });
                    }
                    break;
                    
                case 'clear_door':
                    console.log(`🧹 清空柜门命令: 柜门${command.command.boxId}`);
                    // 更新数据库中的柜门状态为可用(0)，并清空相关信息
                    if (command.command.doorId) {
                        db.run(`UPDATE boxes SET status = 0, phone = '', password = '', updated_at = datetime('now') WHERE id = ?`, 
                               [command.command.doorId], function(err) {
                            if (err) {
                                console.error('❌ 清空柜门失败:', err);
                            } else {
                                console.log(`✅ 柜门${command.command.boxId}已清空`);
                            }
                        });
                    }
                    break;
                    
                case 'open_all_doors':
                    console.log(`🔓 一键开启所有柜门命令`);
                    // 更新所有柜门状态为开启(2)
                    db.run(`UPDATE boxes SET status = 2, updated_at = datetime('now')`, function(err) {
                        if (err) {
                            console.error('❌ 一键开启失败:', err);
                        } else {
                            console.log(`✅ 一键开启成功，影响 ${this.changes} 个柜门`);
                        }
                    });
                    break;
                    
                case 'clear_all_doors':
                    console.log(`🧹 一键清空所有柜门命令`);
                    // 更新所有柜门状态为可用(0)，并清空相关信息
                    db.run(`UPDATE boxes SET status = 0, phone = '', password = '', updated_at = datetime('now')`, function(err) {
                        if (err) {
                            console.error('❌ 一键清空失败:', err);
                        } else {
                            console.log(`✅ 一键清空成功，影响 ${this.changes} 个柜门`);
                        }
                    });
                    break;
                    
                default:
                    console.log(`❓ 未知的柜门命令: ${action}`);
            }
        }
        
    } catch (error) {
        console.error('❌ 解析命令失败:', error);
    }
}

// REST API 路由
// 获取所有设备
app.get('/api/devices', (req, res) => {
    // 查询时合并相同的设备，只显示最新的状态
    db.all(`SELECT 
                device_id,
                name,
                status,
                MAX(last_seen) as last_seen,
                created_at
            FROM devices 
            GROUP BY device_id 
            ORDER BY last_seen DESC`, (err, rows) => {
        if (err) {
            res.status(500).json({ error: err.message });
            return;
        }
        res.json(rows);
    });
});

// 获取消息日志
app.get('/api/messages', (req, res) => {
    const limit = req.query.limit || 100;
    db.all('SELECT * FROM message_logs ORDER BY timestamp DESC LIMIT ?', [limit], (err, rows) => {
        if (err) {
            res.status(500).json({ error: err.message });
            return;
        }
        res.json(rows);
    });
});

// 获取寄存柜数据
app.get('/api/boxes', (req, res) => {
    db.all('SELECT * FROM boxes ORDER BY boxId ASC', (err, rows) => {
        if (err) {
            res.status(500).json({ error: err.message });
            return;
        }
        
        // 如果没有数据，创建一些示例数据
        if (rows.length === 0) {
            const sampleBoxes = [];
            for (let i = 1; i <= 20; i++) {
                sampleBoxes.push({
                    id: i,
                    phone: '',
                    password: '',
                    boxId: i.toString().padStart(3, '0'),
                    size: ['小', '中', '大', '特大'][Math.floor(Math.random() * 4)],
                    status: Math.floor(Math.random() * 3), // 0=可用, 1=已用, 2=开启
                    enabled: true,
                    lockBoard: Math.floor(i / 8) + 1,
                    lockNumber: i,
                    memberFlag: 0
                });
            }
            
            // 插入示例数据
            const stmt = db.prepare(`INSERT OR REPLACE INTO boxes 
                (id, phone, password, boxId, size, status, enabled, lockBoard, lockNumber, memberFlag) 
                VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?)`);
            
            sampleBoxes.forEach(box => {
                stmt.run([box.id, box.phone, box.password, box.boxId, box.size, 
                         box.status, box.enabled, box.lockBoard, box.lockNumber, box.memberFlag]);
            });
            stmt.finalize();
            
            res.json(sampleBoxes);
        } else {
            res.json(rows);
        }
    });
});

// 获取操作记录
app.get('/api/records', (req, res) => {
    const limit = req.query.limit || 100;
    db.all('SELECT * FROM records ORDER BY updated_at DESC LIMIT ?', [limit], (err, rows) => {
        if (err) {
            res.status(500).json({ error: err.message });
            return;
        }
        res.json(rows);
    });
});

// ==================== MQTT HTTP API 接口 ====================

// 获取所有客户端主题列表
app.get('/api/mqtt/client/getTopicList', (req, res) => {
    const status = req.query.status || '';
    
    // 获取所有活跃的订阅
    const subscriptions = [];
    if (aedes.clients) {
        for (let [clientId, client] of Object.entries(aedes.clients)) {
            if (client && client.subscriptions) {
                for (let topic of Object.keys(client.subscriptions)) {
                    if (!topic.startsWith('$SYS/')) {
                        subscriptions.push({
                            clientId: clientId,
                            topic: topic,
                            status: 'online',
                            qos: client.subscriptions[topic].qos || 0
                        });
                    }
                }
            }
        }
    }
    
    res.json({
        success: true,
        data: subscriptions,
        total: subscriptions.length
    });
});

// 获取所有客户端列表
app.get('/api/mqtt/client/getList', (req, res) => {
    const status = req.query.status || '';
    
    const clients = [];
    if (aedes.clients) {
        for (let [clientId, client] of Object.entries(aedes.clients)) {
            clients.push({
                clientId: clientId,
                status: 'online',
                connectedAt: new Date().toISOString(),
                ipAddress: client.conn ? (client.conn.remoteAddress || 'unknown') : 'unknown',
                subscriptions: client.subscriptions ? Object.keys(client.subscriptions).length : 0
            });
        }
    }
    
    res.json({
        success: true,
        data: clients,
        total: clients.length
    });
});

// 获取指定客户端状态
app.get('/api/mqtt/client/getStatus', (req, res) => {
    const clientId = req.query.clientId;
    
    if (!clientId) {
        res.status(400).json({ error: '缺少clientId参数' });
        return;
    }
    
    const isOnline = aedes.clients && aedes.clients[clientId];
    
    res.json({
        success: true,
        data: {
            clientId: clientId,
            status: isOnline ? 'online' : 'offline',
            lastSeen: new Date().toISOString()
        }
    });
});

// 获取指定客户端订阅信息
app.get('/api/mqtt/client/getTopicInfo', (req, res) => {
    const clientId = req.query.clientId;
    
    if (!clientId) {
        res.status(400).json({ error: '缺少clientId参数' });
        return;
    }
    
    const client = aedes.clients ? aedes.clients[clientId] : null;
    const subscriptions = [];
    
    if (client && client.subscriptions) {
        for (let [topic, subscription] of Object.entries(client.subscriptions)) {
            if (!topic.startsWith('$SYS/')) {
                subscriptions.push({
                    topic: topic,
                    qos: subscription.qos || 0,
                    subscribedAt: new Date().toISOString()
                });
            }
        }
    }
    
    res.json({
        success: true,
        data: {
            clientId: clientId,
            subscriptions: subscriptions,
            total: subscriptions.length
        }
    });
});

// 断开指定客户端连接
app.delete('/api/mqtt/client/dellink', (req, res) => {
    const clientId = req.query.clientId;
    
    if (!clientId) {
        res.status(400).json({ error: '缺少clientId参数' });
        return;
    }
    
    const client = aedes.clients ? aedes.clients[clientId] : null;
    
    if (client) {
        client.close();
        res.json({ success: true, message: `客户端 ${clientId} 已断开连接` });
    } else {
        res.status(404).json({ error: '客户端不存在或已离线' });
    }
});

// 发布MQTT消息
app.post('/api/mqtt/message/publish', (req, res) => {
    const { topic, message } = req.body;
    
    if (!topic || message === undefined) {
        res.status(400).json({ error: '缺少topic或message参数' });
        return;
    }
    
    aedes.publish({
        topic: topic,
        payload: typeof message === 'string' ? message : JSON.stringify(message),
        qos: 0
    });
    
    console.log(`📡 HTTP API发布消息: 主题=${topic}, 内容=${message}`);
    
    res.json({ success: true, message: '消息发布成功' });
});

// 订阅主题（内部订阅）
app.post('/api/mqtt/message/subscribe', (req, res) => {
    const { filter, subscription_id } = req.body;
    
    if (!filter) {
        res.status(400).json({ error: '缺少filter参数' });
        return;
    }
    
    // 这里可以实现服务器端订阅逻辑
    res.json({ 
        success: true, 
        message: `已订阅主题: ${filter}`,
        subscription_id: subscription_id || Date.now()
    });
});

// 取消订阅主题
app.post('/api/mqtt/message/unsubscribe', (req, res) => {
    const { filter, subscription_id } = req.body;
    
    if (!filter) {
        res.status(400).json({ error: '缺少filter参数' });
        return;
    }
    
    res.json({ 
        success: true, 
        message: `已取消订阅主题: ${filter}`
    });
});

// 客户端注入（发送特定消息给指定客户端）
app.post('/api/mqtt/client/injection', (req, res) => {
    const { listener, clientId, topicName, payload } = req.body;
    
    if (!clientId || !topicName) {
        res.status(400).json({ error: '缺少clientId或topicName参数' });
        return;
    }
    
    // 发布消息到指定主题（客户端如果订阅了会收到）
    aedes.publish({
        topic: topicName,
        payload: payload || '',
        qos: 0
    });
    
    console.log(`💉 客户端注入: 目标=${clientId}, 主题=${topicName}, 内容=${payload}`);
    
    res.json({ 
        success: true, 
        message: `已向客户端 ${clientId} 发送消息`
    });
});

// 获取访问Token（简单实现）
app.post('/api/mqtt/accessToken', (req, res) => {
    const { secret } = req.body;
    
    // 简单的token生成（生产环境需要更安全的实现）
    const token = Buffer.from(`mqtt_${Date.now()}_${Math.random()}`).toString('base64');
    
    res.json({
        success: true,
        data: {
            token: token,
            expiresIn: 3600, // 1小时
            scope: 'mqtt_management'
        }
    });
});

// 兼容原有的发布接口
app.post('/api/publish', (req, res) => {
    const { topic, payload } = req.body;
    
    if (!topic || !payload) {
        res.status(400).json({ error: '缺少topic或payload参数' });
        return;
    }
    
    aedes.publish({
        topic: topic,
        payload: JSON.stringify(payload)
    });
    
    res.json({ success: true, message: '消息发布成功' });
});

// 更新柜门状态
app.post('/api/boxes/:id/status', (req, res) => {
    const boxId = req.params.id;
    const { status } = req.body;
    
    if (status === undefined) {
        res.status(400).json({ error: '缺少status参数' });
        return;
    }
    
    db.run(`UPDATE boxes SET status = ?, updated_at = datetime('now') WHERE id = ?`, 
           [status, boxId], function(err) {
        if (err) {
            res.status(500).json({ error: err.message });
            return;
        }
        
        if (this.changes === 0) {
            res.status(404).json({ error: '柜门不存在' });
            return;
        }
        
        res.json({ success: true, message: '柜门状态更新成功' });
    });
});

// 批量更新柜门状态
app.post('/api/boxes/batch-update', (req, res) => {
    const { status, action } = req.body;
    
    if (status === undefined) {
        res.status(400).json({ error: '缺少status参数' });
        return;
    }
    
    db.run(`UPDATE boxes SET status = ?, updated_at = datetime('now')`, 
           [status], function(err) {
        if (err) {
            res.status(500).json({ error: err.message });
            return;
        }
        
        res.json({ 
            success: true, 
            message: `批量操作成功，影响 ${this.changes} 个柜门`,
            changes: this.changes 
        });
    });
});

// WebSocket连接处理
wss.on('connection', (ws) => {
    console.log('🌐 Web客户端连接');
    console.log(`📊 当前Web客户端数量: ${wss.clients.size}`);
    
    // 发送当前连接状态给新连接的客户端
    ws.send(JSON.stringify({
        type: 'connection_status',
        message: 'WebSocket连接成功',
        timestamp: moment().format('YYYY-MM-DD HH:mm:ss')
    }));
    
    ws.on('message', (message) => {
        try {
            const data = JSON.parse(message);
            console.log('📩 收到Web客户端消息:', data);
            
            // 处理Web客户端的请求
            if (data.type === 'subscribe') {
                // 可以实现特定的订阅逻辑
            }
        } catch (error) {
            console.error('❌ 处理Web客户端消息失败:', error);
        }
    });
    
    ws.on('close', () => {
        console.log('🌐 Web客户端断开连接');
        console.log(`📊 当前Web客户端数量: ${wss.clients.size - 1}`);
    });
});

// 启动服务器
function startServer() {
    // 初始化数据库
    initDatabase();
    
    // 启动MQTT服务器
    server.listen(MQTT_PORT, '0.0.0.0', () => {
        console.log(`🚀 MQTT服务器启动在端口 ${MQTT_PORT}`);
        console.log(`🔗 MQTT服务器监听地址: 0.0.0.0:${MQTT_PORT}`);
    });
    
    // 启动Web服务器
    app.listen(WEB_PORT, () => {
        console.log(`🌐 Web服务器启动在端口 ${WEB_PORT}`);
        console.log(`📱 管理界面: http://localhost:${WEB_PORT}`);
    });
    
    console.log(`🔌 WebSocket服务器启动在端口 ${WS_PORT}`);
    console.log('✅ MQTT服务器框架启动完成！');
    console.log('');
    console.log('📋 服务状态:');
    console.log(`   - MQTT服务器: localhost:${MQTT_PORT}`);
    console.log(`   - Web管理界面: http://localhost:${WEB_PORT}`);
    console.log(`   - WebSocket服务: ws://localhost:${WS_PORT}`);
    console.log('');
    console.log('🎯 等待客户端连接...');
    
    // 启动心跳测试机制
    startHeartbeatTest();
}

// 添加心跳测试机制
function startHeartbeatTest() {
    setInterval(() => {
        // 检查是否有连接的客户端
        const clientCount = aedes.clients ? Object.keys(aedes.clients).length : 0;
        if (clientCount > 0) {
            // 减少日志输出，只在30秒时输出一次详细信息
            const now = new Date();
            const shouldLog = now.getSeconds() === 0 || now.getSeconds() === 30;
            
            if (shouldLog) {
                console.log(`💓 心跳检测: ${clientCount} 个客户端在线`);
            }
            
            aedes.publish({
                topic: 'heart',
                payload: JSON.stringify({
                    type: 'server_heartbeat',
                    timestamp: moment().format(),
                    message: '服务器心跳测试'
                })
            });
        }
    }, 30000); // 改为每30秒发送一次
}

// 优雅关闭
process.on('SIGINT', () => {
    console.log('\n正在关闭服务器...');
    db.close();
    process.exit(0);
});

// 启动服务器
startServer();